package com.flink.timewindow.process;

import com.flink.timewindow.bean.WaterSensor;
import com.flink.timewindow.function.WaterSensorMapFunction;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class KeyedProcessTimerDemo {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        //获取数据源
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("10.90.100.102", 8888)
                //数据处理
                //切分转换
                .map(new WaterSensorMapFunction());

        //TODO 指定Watermark策略
          //定义watermark策略
        WatermarkStrategy<WaterSensor> waterSensorWatermarkStrategy = WatermarkStrategy
                //指定watermark生成：乱序的，有等待时间：等待三秒
                .<WaterSensor>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                //指定时间戳分配器，从数据中提取
                .withTimestampAssigner(new SerializableTimestampAssigner<WaterSensor>() {
                    @Override
                    public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                        //返回的时间戳需要毫秒，所以得乘1000
                        return element.getTs()*1000L;
                    }
                });
        SingleOutputStreamOperator<WaterSensor> sensorDSwithWatermark = sensorDS.assignTimestampsAndWatermarks(waterSensorWatermarkStrategy);


        //分组
        KeyedStream<WaterSensor, String> sensorKS = sensorDSwithWatermark.keyBy(value -> value.getId());


        //TODO
        SingleOutputStreamOperator<String> process = sensorKS.process(
                                       //key的类型   输入数据的类型   输出类型
                new KeyedProcessFunction<String, WaterSensor, String>() {
                    /**
                     * 来一条数据调用一次
                     * @param value 输入的数据
                     * @param ctx   上下文.
                     * @param out   收集器.
                     * @throws Exception
                     */
                   @Override
                  public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                        //数据中提取出来的事件时间
                    Long currentEventTime = ctx.timestamp();
                    //定时器
                    TimerService timerService = ctx.timerService();
                    //注册定时器： 事件时间
                    timerService.registerEventTimeTimer(5000L); //定义一个5秒的定时器，时间到5秒了就该触发到onTimer方法
                     System.out.println("key为"+ctx.getCurrentKey()+"当前时间是"+currentEventTime+",注册了一个5秒的定时器");
                       //注册定时器：处理时间
                    //timerService.registerProcessingTimeTimer();
                    //删除定时器： 事件时间
                   // timerService.deleteEventTimeTimer();
                    //删除定时器： 处理时间
                    //timerService.deleteProcessingTimeTimer();
                    //当前处理时间即当前系统时间
                  //  long currentTS = timerService.currentProcessingTime();
                    //当前水位线
                    //long wm = timerService.currentWatermark();



                       /**
                        //处理时间案例
                        long currentProcessingTime = timerService.currentProcessingTime();
                        timerService.registerProcessingTimeTimer(currentProcessingTime+5000L);
                        System.out.println("当前key="+ctx.getCurrentKey()+", 当前时间="+currentProcessingTime+",注册了一个5秒后的定时器");
                        **/

                       /** **/
                        //获取process的当前watermark案例
                        long currentWatermark = timerService.currentWatermark();
                        System.out.println("当前数据="+value+", 当前watermark="+currentWatermark);


                  }

                    /**
                     * 时间进展到定时器注册的时间，调用该方法
                     * @param timestamp 当前时间进展
                     * @param ctx       上下文
                     * @param out      采集器
                     * @throws Exception
                     */
                    @Override
                   public void onTimer(long timestamp, KeyedProcessFunction<String, WaterSensor, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                        super.onTimer(timestamp,ctx,out);
                        System.out.println("当前的key为："+ctx.getCurrentKey()+"现在时间是"+timestamp+"定时器触发");



                   }

                });




        //执行
        env.execute();

    }
}
